/*
 * Copyright 2013-2017 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.springframework.cloud.gateway.filter.factory;

import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandKey;
import com.netflix.hystrix.HystrixObservableCommand;
import com.netflix.hystrix.exception.HystrixRuntimeException;
import org.springframework.cloud.gateway.filter.GatewayFilter;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.http.HttpStatus;
import org.springframework.tuple.Tuple;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import rx.Observable;
import rx.RxReactiveStreams;
import rx.Subscription;

import java.util.Arrays;
import java.util.List;
import java.util.function.Function;

import static com.netflix.hystrix.exception.HystrixRuntimeException.FailureType.TIMEOUT;
import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.setResponseStatus;

/**
 * @author Spencer Gibb
 */
public class HystrixGatewayFilterFactory implements GatewayFilterFactory {

	@Override
	// 定义了Tuple参数的Key为name
	public List<String> argNames() {
		return Arrays.asList(NAME_KEY);
	}

	@Override
	// 创建HystrixGatewayFilter对象
	public GatewayFilter apply(Tuple args) {
		// TODO: if no name is supplied, generate one from command id (useful for default filter)
		// 从Tuple参数获得Hystrix Command名字，例如上面举例的RouteDefinition时，commandName=myCommandName
		final String commandName = args.getString(NAME_KEY);
		// 创建Hystrix Command分组Key为HystrixGatewayFilterFactory
		final HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey(getClass().getSimpleName());
		// 创建Hystrix Command Key为commandName
		final HystrixCommandKey commandKey = HystrixCommandKey.Factory.asKey(commandName);
		
		// 创建HystrixObservableCommand.Setter对象
		final HystrixObservableCommand.Setter setter = HystrixObservableCommand.Setter
				.withGroupKey(groupKey)
				.andCommandKey(commandKey);

		// 创建HystrixGatewayFilter对象并返回
		return (exchange, chain) -> {
			// 创建RouteHystrixCommand对象
			RouteHystrixCommand command = new RouteHystrixCommand(setter, exchange, chain);
			
			// 调用Mono#create(Consumer<MonoSink<T>>)方法，创建Mono对象，因为Hystrix基于RxJava，而GatewayFilter基于Reactor（Mono是其内部的一个类），通过这个方法，实现订阅的适配
			return Mono.create(s -> {
			    // 1）调用RouteHystrixCommand#toObservable()方法，内部会调用RouteHystrixCommand#construct()方法，获得执行this.chain.filter(this.exchange)的Observable
				// 2）订阅Observable：成功或完成时，调用Mono#success(Object)方法，目前创建的Mono上没有相关的订阅；异常时，调用Mono#error(Object)方法，目前创建的Mono上调用Mongo#onErrorResume(Function<Throwable, Mono<Void>>))方法，进行订阅
				Subscription sub = command.toObservable().subscribe(s::success, s::error, s::success);
				// Mono取消时，取消Hystrix Command Observable的订阅，结束Hystrix Command的执行
				s.onCancel(sub::unsubscribe);
			// 当Hystrix Command执行超时时，设置响应504状态码，并回写客户端响应（exchange.getResponse().setComplete()）
			// 注意：目前Hystrix Command执行超时时，返回客户端504状态码，如果使用JSON格式作为数据返回，则需要修改下该HystrixGatewayFilter的代码实现
			}).onErrorResume((Function<Throwable, Mono<Void>>) throwable -> {
				if (throwable instanceof HystrixRuntimeException) {
					HystrixRuntimeException e = (HystrixRuntimeException) throwable;
					if (e.getFailureType() == TIMEOUT) { //TODO: optionally set status
						setResponseStatus(exchange, HttpStatus.GATEWAY_TIMEOUT);
						return exchange.getResponse().setComplete();
					}
				}
				// 当Hystrix Command发生其他异常时，例如断路器打开，返回Mono.empty()，最终返回客户端200状态码，内容为空
				// 注意：当Hystrix熔断时，最终返回客户端200状态码，内容为空，此处建议改HystrixGatewayFilter的代码实现
				return Mono.empty();
			}).then(); // 调用Mono#then()方法，参数为空，返回空Mono，不再向后发射数据
		};
	}

	//TODO: replace with HystrixMonoCommand that we write
	private class RouteHystrixCommand extends HystrixObservableCommand<Void> {
		private final ServerWebExchange exchange;
		private final GatewayFilterChain chain;

		RouteHystrixCommand(Setter setter, ServerWebExchange exchange, GatewayFilterChain chain) {
			super(setter);
			this.exchange = exchange;
			this.chain = chain;
		}

		@Override
		protected Observable<Void> construct() {
			return RxReactiveStreams.toObservable(this.chain.filter(this.exchange));
		}
	}
}
